package com.qubi.exchange.websocket;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.qubi.exchange.biz.ColaExchangeBiz;
import com.qubi.exchange.biz.ColaSymbolBiz;
import com.qubi.exchange.biz.ExchangeRate;
import com.qubi.exchange.config.SpringBeanHandler;
import com.qubi.exchange.constant.MessageType;
import com.qubi.exchange.message.BufferMessage;
import com.qubi.exchange.message.NotifyMessage;
import com.qubi.exchange.protobuffer.BufferTransfrom;
import com.qubi.exchange.protobuffer.DepthBuffer;
import com.qubi.exchange.protobuffer.KlinesBuffer;
import com.qubi.exchange.protobuffer.MarketBuffer;
import com.qubi.exchange.service.MatchService;
import com.qubi.exchange.util.GZipUtils;
import com.qubi.me.entity.ColaCoinSymbol;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
@ServerEndpoint("/webSocket/v2")
@Log4j2
public class WebSocketService {

    /**
     * 订阅主题
     */
    public static Map<Session, List<String>> sessionTopics = new ConcurrentHashMap<>();

    /**
     * 容器
     */
    public static Map<String, ConcurrentHashMap<Session, Integer>> topicSession = new ConcurrentHashMap<>();

    /**
     * 服务线程数
     */
    ExecutorService service = Executors.newFixedThreadPool(200);

    public static final String KLINE = "kline";

    public static final String SUBSCRIBE = "subscribe";

    public static final String TYPE = "type";

    public static final String PING = "ping";

    public static final String FROM = "from";

    public static final String TO = "to";

    /**
     * 请求
     */
    public static final String REQ = "req";

    public static final String OK = "ok";

    public static final String STATE = "subscribeState";

    /**
     * 全局数据
     */
    public static final String ALL = "_all";


    public static final String MARKETALL = "market_all";

    public static final String DEPTH = "depth";
    public static final String ORDER="order";


    public final static int MINGZIPLEN=180;

    @OnOpen
    public void onOpen(Session session) {
    }

    @OnError
    public void onError(Session session, Throwable error) {
        clearTopic(session);
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose(Session session) {
        clearTopic(session);
    }


    static Timer timer=new Timer();
    static {
        timer.schedule(new TimerTask() {
            @Override
            public void run() {

                for(Session session:sessionTopics.keySet()){

                    synchronized (session) {
                        if (!session.isOpen()) {
                            try {
                                session.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                            clearTopic(session);
                        }
                    }

                }
                log.info("清除已关闭session");
                log.info("session个数:"+sessionTopics.size());

            }
        },10000,60*1000);
    }
    /**
     * 收到客户端的消息
     *
     * @param message 消息
     * @param session 会话
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        // 消息内容
        JSONObject jsonObject = JSONObject.parseObject(message);
        String type = jsonObject.getString(TYPE);
        if (StrUtil.isEmpty(type)) {
            type = SUBSCRIBE;
        }
        log.info("收到订阅: " + message);
        switch (type) {
            case SUBSCRIBE: {
                var subscribes = jsonObject.getJSONArray(SUBSCRIBE);
                var sessionSubscribes = sessionTopics.getOrDefault(session, new ArrayList<>());
                for (String t : sessionSubscribes) {
                    var topicMap = topicSession.get(t);
                    if (t != null) topicMap.remove(session);

                }
                sessionSubscribes.clear();
                for (Object object : subscribes) {
                    String key = object.toString();
                    ConcurrentHashMap<Session, Integer> hashSet = topicSession.getOrDefault(key, new ConcurrentHashMap<>());
                    hashSet.put(session, 1);
                    sessionSubscribes.add(key);
                    if (key.contains(KLINE)) {
                        this.sendFirstKline(key, session);

                    } else if (key.contains(MARKETALL)) {
                        this.sendMarket(session);

                    } else if (key.contains(DEPTH)) {
                        sendDepth(session, key);
                    }else if(key.contains(ORDER))
                    {
                        sendTrade(session,key);
                    }
                    topicSession.put(key, hashSet);
                }
                session.setMaxBinaryMessageBufferSize(1024*8*100);

                sessionTopics.put(session, sessionSubscribes);
                SpringBeanHandler.getBean(NotifyMessageContainer.class).putContainer(new NotifyMessageContainer.SessonMessageContainer(session));
                sendBufferMessageToSession(session, new NotifyMessage(SUBSCRIBE,"ok"));

                break;
            }
            case PING: {
                sendBufferMessageToSession(session, new NotifyMessage(PING,""));
                break;
            }
            case REQ: {
                Long from = jsonObject.getLong(FROM);

                Long to = jsonObject.getLong(TO);

                String key = jsonObject.getString(SUBSCRIBE);
                if (ObjectUtil.isNotEmpty(from) && ObjectUtil.isNotEmpty(to)) {
                    this.sendKline(from, to, key, session);
                }
                break;
            }
        }
    }

    /**
     * 发送历史kline数据
     *
     * @param from
     * @param to
     * @param topic
     * @param session
     */
    public void sendKline(Long from, Long to, String topic, Session session) {
        try {
            String[] split = StrUtil.split(topic, StrUtil.UNDERLINE);
            // 交易对
            String pair = StrUtil.concat(Boolean.FALSE, split[1], StrUtil.UNDERLINE, split[2]);
            // 分辨率
            String klineType = split[3];
            List<Number[]> kline = SpringBeanHandler.getBean(ColaExchangeBiz.class)
                    .kline(pair, from, to, 800, klineType);
            var data = BufferTransfrom.KlinesTransform.getBuffer(kline, StrUtil.concat(Boolean.FALSE, topic, ALL));
            // 开始发送消息
            sendBufferMessageToSession(session, new NotifyMessage(topic,data));
        } catch (Exception e) {
            log.error("kline数据出错", e);
        }
    }

    /**
     * 第一次Kline数据
     *
     * @param topic
     * @param session
     */
    public void sendFirstKline(String topic, Session session) {
        try {
            String[] split = StrUtil.split(topic, StrUtil.UNDERLINE);
            // 交易对
            String pair = StrUtil.concat(Boolean.FALSE, split[1], StrUtil.UNDERLINE, split[2]);
            // 分辨率
            String klineType = split[3];
            List<Number[]> kline = SpringBeanHandler.getBean(ColaExchangeBiz.class)
                    .kline(pair, null, null, 800, klineType);

            sendBufferMessageToSession(session, new NotifyMessage(topic,kline));

            // 开始发送消息
        } catch (Exception e) {
            log.error("kline数据出错", e);
        }
    }


    public void sendDepth(Session session, String key) {
        String[] split = StrUtil.split(key, StrUtil.UNDERLINE);
        String pair = StrUtil.concat(Boolean.FALSE, split[1], StrUtil.UNDERLINE, split[2]);
        int scale = 3;
        int length = 2000;
        ColaCoinSymbol colaCoinSymbol = SpringBeanHandler.getBean(ColaSymbolBiz.class).getPair(pair);
        scale = colaCoinSymbol.getPriceScale();
        List<BigDecimal[]> ask = MatchService.sellDepth.get(pair).getDepth(length, scale);
        Collections.reverse(ask);
        List<BigDecimal[]> bids = MatchService.buyDepth.get(pair).getDepth(length, scale);
        var current = SpringBeanHandler.getBean(ColaExchangeBiz.class).getMarketByPair(pair, null);
        Map<String,Object>objectMap=new HashMap<>(5);
        objectMap.put("current",current);
        objectMap.put("bid",bids);
        objectMap.put("ask",ask);
        sendBufferMessageToSession(session, new NotifyMessage(key,objectMap));



    }


    public void sendMarket(Session session) {
        var symbols = SpringBeanHandler.getBean(ColaSymbolBiz.class).getCoinSymbolBySymbol(null);
        var markets = SpringBeanHandler.getBean(ColaExchangeBiz.class).getMarketAll(null);
        Map<String,Object>objectMap=new HashMap<>(6);
        var rate= ExchangeRate.currency();
        objectMap.put("areas",SpringBeanHandler.getBean(ColaSymbolBiz.class).getSymbols());
        objectMap.put("indexArea",SpringBeanHandler.getBean(ColaSymbolBiz.class).getIndexSymbols());

        objectMap.put("symbols",symbols);
        objectMap.put("markets",markets);
        objectMap.put("rate",rate);
        sendBufferMessageToSession(session, new NotifyMessage(MARKETALL,objectMap));
    }

    public void sendTrade(Session session, String key) {
        String[] split = StrUtil.split(key, StrUtil.UNDERLINE);
        String pair = StrUtil.concat(Boolean.FALSE, split[1], StrUtil.UNDERLINE, split[2]);
      var data=  SpringBeanHandler.getBean(ColaExchangeBiz.class).getMarketOrder(pair,30);
      sendBufferMessageToSession(session, new NotifyMessage(key,data));
    }



    public void sendBufferMessageToSession(Session session, NotifyMessage notifyMessage) {

       SpringBeanHandler.getBean(NotifyMessageContainer.class).putMessage(session,notifyMessage);
    }


    public void sendMessageToTopic(NotifyMessage notifyMessage) {
        String topic = notifyMessage.getTopic();
        sendMessages(topic, notifyMessage);
    }

    private void sendMessages(String topic, NotifyMessage notifyMessage) {
        var sessions = topicSession.get(topic);
        if (sessions != null) {
            for (Session session : sessions.keySet()) {
                SpringBeanHandler.getBean(NotifyMessageContainer.class).putMessage(session,notifyMessage);

            }
        }
    }

    /**
     * 移除订阅
     *
     * @param session
     */
    public  static void clearTopic(Session session){
        var topics = sessionTopics.get(session);
        SpringBeanHandler.getBean(NotifyMessageContainer.class).remove(session);
        if (topics != null) {
            for (String topic : topics) {
                var data=topicSession.get(topic);
                if(data!=null) data.remove(session);
            }
            topics.clear();
            sessionTopics.remove(session);
        }
    }

}
